package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggest;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.alibaba.csp.sentinel.util.AssertUtil;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

    @Qualifier("com.atguigu.tingshu.album.client.AlbumInfoFeignClient")
    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;
    @Qualifier("com.atguigu.tingshu.album.client.CategoryFeignClient")
    @Autowired
    private CategoryFeignClient categoryFeignClient;

    @Qualifier("com.atguigu.tingshu.user.client.UserInfoFeignClient")
    @Autowired
    private UserInfoFeignClient userInfoFeignClient;
    @Autowired
    private AlbumIndexRepository albumIndexRepository;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;
    @Autowired
    private ElasticsearchClient elasticsearchClient;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedissonClient redissonClient;

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String dimension) {
        return (List<AlbumInfoIndexVo>) redisTemplate.boundHashOps(RedisConstant.RANKING_KEY_PREFIX + category1Id).get(dimension);
    }

    @Override
    public void updateLatelyAlbumRanking() {
        Result<List<BaseCategory1>> category1 = categoryFeignClient.findAllCategory1();
        AssertUtil.notNull(category1, "category1 is null");
        List<BaseCategory1> category = category1.getData();
        Assert.notNull(category, "category is null");
        category.forEach(baseCategory1 -> {
            String[] rankingDimensionArray = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
            for (String ranging : rankingDimensionArray) {
                SearchRequest.Builder builder = new SearchRequest.Builder();
                builder.index("albuminfo")
                        .query(query -> query.term(match -> match.field("category1Id").value(baseCategory1.getId())))
                        .sort(sort -> sort.field(field -> field.field(ranging).order(SortOrder.Desc)))
                        .size(10);
                SearchRequest build = builder.build();
                SearchResponse<AlbumInfoIndex> response = null;
                try {
                    response=elasticsearchClient.search(build, AlbumInfoIndex.class);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                List<AlbumInfoIndex> albumInfoIndexList = response.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
                //  使用hash 数据结构 hset key field value;
                String rangKey = RedisConstant.RANKING_KEY_PREFIX+baseCategory1.getId();
                this.redisTemplate.boundHashOps(rangKey).put(ranging,albumInfoIndexList);
            }
        });
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index("suggestinfo").suggest(suggest -> suggest.suggesters("suggestionKeyword", s -> s.prefix(keyword)
                .completion(c -> c.field("keyword").skipDuplicates(true).size(10).fuzzy(f -> f.fuzziness("AUTO"))))
        ).suggest(suggest -> suggest.suggesters("suggestionkeywordPinyin", s1 -> s1.prefix(keyword)
                .completion(c -> c.field("keywordPinyin").skipDuplicates(true).size(10).fuzzy(f -> f.fuzziness("AUTO"))))
        ).suggest(suggest -> suggest.suggesters("suggestionkeywordSequence", s1 -> s1.prefix(keyword)
                .completion(c -> c.field("keywordSequence").skipDuplicates(true).size(10).fuzzy(f -> f.fuzziness("AUTO")))));
        SearchResponse<SuggestIndex> search = null;
        try {
            search = elasticsearchClient.search(builder.build(), SuggestIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        HashSet<String> titleSet = new HashSet<>();
        titleSet.addAll(this.parseResultData(search, "suggestionKeyword"));
        titleSet.addAll(this.parseResultData(search, "suggestionkeywordPinyin"));
        titleSet.addAll(this.parseResultData(search, "suggestionkeywordSequence"));
        if (titleSet.size() < 12) {
            SearchResponse<SuggestIndex> response = null;
            try {
                response = elasticsearchClient.search(s -> s.index("suggestinfo")
                                .query(f -> f.match(m -> m.field("title").query(keyword)))
                        , SuggestIndex.class);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            //  从查询结果集中获取数据
            for (Hit<SuggestIndex> hit : response.hits().hits()) {
                //  获取数据结果
                SuggestIndex suggestIndex = hit.source();
                //  获取titile
                titleSet.add(suggestIndex.getTitle());
                //  判断当前这个结合的长度.
                if (titleSet.size() == 12) {
                    break;
                }
            }
        }
        return new ArrayList<>(titleSet);
    }

    private Collection<String> parseResultData(SearchResponse<SuggestIndex> search, String suggestionKeyword) {
        List<String> suggestList = new ArrayList<>();
        Map<String, List<Suggestion<SuggestIndex>>> groupBySuggestionListAggMap = search.suggest();
        if(groupBySuggestionListAggMap.get(suggestionKeyword)!=null){
            groupBySuggestionListAggMap.get(suggestionKeyword).forEach(item -> {
                CompletionSuggest<SuggestIndex> completionSuggest = item.completion();
                completionSuggest.options().forEach(it -> {
                    SuggestIndex suggestIndex = it.source();
                    suggestList.add(suggestIndex.getTitle());
                });
            });
            //  返回集合列表
        }
        return suggestList;
    }

    @Override
    public List<Map<String, Object>> channel(Long category1Id) throws IOException {
        Result<List<BaseCategory3>> topBaseCategory3 = categoryFeignClient.findTopBaseCategory3(category1Id);
        AssertUtil.notNull(topBaseCategory3, "三级分类数据集查询失败");
        List<BaseCategory3> baseCategory3Data = topBaseCategory3.getData();
        AssertUtil.notNull(baseCategory3Data, "三级分类查询失败");
        //Map<String, Object> map = new HashMap<>();
        Map<Long, BaseCategory3> baseCategory3Map = baseCategory3Data.stream()
                .collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));
        List<Long> idList = baseCategory3Data.stream().map(BaseCategory3::getId).collect(Collectors.toList());
        List<FieldValue> valueList = idList.stream().map(id -> FieldValue.of(id)).collect(Collectors.toList());

        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index("albuminfo").query(q -> q.terms(t -> t.field("category3Id")
                        .terms(ts -> ts.value(valueList))))
                .aggregations("groupByCategory3IdAgg", a -> a.terms(t -> t.field("category3Id").size(10))
                        .aggregations("topSixHotScoreAgg", ag -> ag.topHits(th -> th.size(6)
                                .sort(s -> s.field(f -> f.field("hotScore").order(SortOrder.Desc))))));
        SearchResponse<AlbumInfoIndex> search = elasticsearchClient.search(builder.build(), AlbumInfoIndex.class);
        List<Map<String, Object>> list = new ArrayList<>();
        Aggregate groupByCategory3IdAgg = search.aggregations().get("groupByCategory3IdAgg");
        if (groupByCategory3IdAgg != null) {
            LongTermsBucket bucket = groupByCategory3IdAgg.lterms().buckets().array().get(0);
            Map<String, Object> map = new HashMap<>();
            List<AlbumInfoIndex> albumInfoIndexList = new ArrayList<>();
            long category3Id = bucket.key();
            Aggregate topTenHotScoreAgg = bucket.aggregations().get("topSixHotScoreAgg");
            if (topTenHotScoreAgg != null) {
                topTenHotScoreAgg.topHits().hits().hits().forEach(hit -> {
                    String s = hit.source().toString();
                    AlbumInfoIndex albumInfoIndex = JSON.parseObject(s, AlbumInfoIndex.class);
                    albumInfoIndexList.add(albumInfoIndex);
                });
            }
            map.put("baseCategory3", baseCategory3Map.get(category3Id));
            map.put("list", albumInfoIndexList);
            list.add(map);


        }
        return list;
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) throws IOException {
        SearchRequest request = this.buildQueryDsl(albumIndexQuery);
        SearchResponse<AlbumInfoIndex> albumInfoIndexSearchResponse = elasticsearchClient.search(request, AlbumInfoIndex.class);
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        if (albumInfoIndexSearchResponse.hits().hits().size() > 0) {
            albumSearchResponseVo.setTotal(albumInfoIndexSearchResponse.hits().total().value());
            albumSearchResponseVo.setPageNo(albumIndexQuery.getPageNo());
            albumSearchResponseVo.setPageSize(albumIndexQuery.getPageSize());
            albumSearchResponseVo.setList(albumInfoIndexSearchResponse.hits().hits().stream().map(hit -> {
                AlbumInfoIndex albumInfoIndex = hit.source();
                AlbumInfoIndexVo albumInfoVo = new AlbumInfoIndexVo();
                BeanUtils.copyProperties(albumInfoIndex, albumInfoVo);
                if (null != hit.highlight().get("albumTitle")) {
                    //  获取高亮数据
                    String albumTitle = hit.highlight().get("albumTitle").get(0);
                    //  赋值高亮数据
                    albumInfoVo.setAlbumTitle(albumTitle);
                }
                return albumInfoVo;
            }).collect(Collectors.toList()));
            albumSearchResponseVo.setTotalPages((long) Math.ceil(albumSearchResponseVo.getTotal() * 1.0 / albumIndexQuery.getPageSize()));

        }
        return albumSearchResponseVo;
    }

    private SearchRequest buildQueryDsl(AlbumIndexQuery albumIndexQuery) {
        String keyword = albumIndexQuery.getKeyword();
        SearchRequest.Builder builder = new SearchRequest.Builder();

        if (StringUtils.hasText(keyword)) {
            SearchRequest.Builder query = builder.index("albuminfo").query(q -> q.bool(b -> b.should(s -> s.match(m -> m.field("albumTitle").query(keyword)))
                    .should(sh -> sh.match(m -> m.field("albumIntro").query(keyword)))));
            builder.highlight(h -> h.fields("albumTitle", f -> f.preTags("<font color='red'>").postTags("</font>")));
        }
        if (albumIndexQuery.getCategory1Id() != null && albumIndexQuery.getCategory1Id() != 0) {
            SearchRequest.Builder query = builder.index("albuminfo").query(q ->
                    q.bool(b -> b.filter(f -> f.term(t -> t.field("category1Id").value(albumIndexQuery.getCategory1Id())))));

        }
        if (albumIndexQuery.getCategory2Id() != null && albumIndexQuery.getCategory2Id() != 0) {
            builder.index("albuminfo").query(q -> q.bool(b -> b.filter(f -> f.term(t -> t.field("category2Id").value(albumIndexQuery.getCategory2Id())))));

        }
        if (albumIndexQuery.getCategory3Id() != null && albumIndexQuery.getCategory3Id() != 0) {
            builder.index("albuminfo").query(q -> q.bool(b -> b.filter(f -> f.term(t -> t.field("category3Id").value(albumIndexQuery.getCategory3Id())))));

        }
        if (albumIndexQuery.getAttributeList() != null) {
            albumIndexQuery.getAttributeList().forEach(attribute -> {
                String[] split = attribute.split(":");
                if (split.length == 2) {
                    String attributeId = split[0];
                    String attributeValueId = split[1];
                    builder.index("albuminfo").query(q -> q.bool(b -> b.filter(
                            f -> f.nested(n -> n.path("attributeValueIndexList").
                                    query(qu -> qu.bool(bo -> bo.must(m -> m.term(
                                                    t -> t.field("attributeValueIndexList.attributeId").value(attributeId)))
                                            .must(mu -> mu.term(ter -> ter.field("attributeValueIndexList.valueId").value(attributeValueId)))))))));
                }
            });
        }
        if (StringUtils.hasText(albumIndexQuery.getOrder())) {
            String[] split = albumIndexQuery.getOrder().split(":");
            if (split.length == 2) {
                switch (split[0]) {
                    case "2":
                        builder.index("albuminfo").sort(s -> s.field(f -> f.field("playStatNum")
                                .order(split[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc)));
                        break;
                    case "3":
                        builder.index("albuminfo").sort(
                                s -> s.field(f -> f.field("createTime").
                                        order(split[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc)));
                        break;
                    case "1":
                        builder.index("albuminfo").sort(
                                s -> s.field(f -> f.field("hotScore").
                                        order(split[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc)));
                        break;
                    default:
                        builder.index("albuminfo").sort(
                                s -> s.field(f -> f.field("_score").
                                        order(SortOrder.Desc)));
                }

            }

        }else{
            builder.index("albuminfo").sort(
                    s -> s.field(f -> f.field("_score").
                            order(SortOrder.Desc)));
        }
        builder.source(s -> s.filter(f -> f.excludes("attributeValueIndexList")));
        Integer from = (albumIndexQuery.getPageNo() - 1) * albumIndexQuery.getPageSize();
        builder.from(from);
        builder.size(albumIndexQuery.getPageSize());


        SearchRequest build = builder.build();
        System.out.println("dsl:\t" + build.toString());
        return build;
    }

    @Override
    public void lowerAlbum(Long albumId) {
        albumIndexRepository.deleteById(albumId);
    }

    @Override
    public void upperAlbum(Long albumId) {
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
       // AlbumInfo albumInfo=new AlbumInfo();
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            AlbumInfo albumInfo = albumInfoFeignClient.getAlbumInfo(albumId).getData();//根据专辑id查询专辑信息
            if (albumInfo != null) {
                BeanUtils.copyProperties(albumInfo, albumInfoIndex);
            }
            return albumInfo;
        });
        CompletableFuture<Void> statCompletableFuture = CompletableFuture.runAsync(() -> {
            AlbumStatVo albumStatVo = albumInfoFeignClient.getAlbumStatVo(albumId).getData();
            if (albumStatVo != null) {
                BeanUtils.copyProperties(albumStatVo, albumInfoIndex);
            }
        });

        CompletableFuture<Void> categoryCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            BaseCategoryView baseCategoryView = categoryFeignClient.getCategoryView(albumInfo.getCategory3Id()).getData();
            if (baseCategoryView != null) {
                BeanUtils.copyProperties(baseCategoryView, albumInfoIndex);
            }
        });

        CompletableFuture<Void> userCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<UserInfoVo> userInfoVo = userInfoFeignClient.getUserInfoVo(albumInfo.getUserId());
            AssertUtil.notNull(userInfoVo, "用户信息为空");
            if (userInfoVo.getData() != null) {
                albumInfoIndex.setAnnouncerName(userInfoVo.getData().getNickname());
            }
        });

        CompletableFuture<Void> attributeCompletableFuture = CompletableFuture.runAsync(() -> {
            List<AlbumAttributeValue> albumAttributeValues = albumInfoFeignClient.findAlbumAttributeValue(albumId).getData();
            if (albumAttributeValues != null) {
                List<AttributeValueIndex> attributeValueIndices = albumAttributeValues.stream().map(albumAttributeValue -> {
                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                    BeanUtils.copyProperties(albumAttributeValue, attributeValueIndex);
                    return attributeValueIndex;
                }).collect(Collectors.toList());
                albumInfoIndex.setAttributeValueIndexList(attributeValueIndices);
            }
        });
        CompletableFuture<Void> keyWordCompletableFuture = CompletableFuture.runAsync(() -> {
            //  创建对象 专辑标题提词
            SuggestIndex suggestIndex = new SuggestIndex();
            suggestIndex.setId(UUID.randomUUID().toString().replaceAll("-", ""));
            suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
            suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
            suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
            suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));
            this.suggestIndexRepository.save(suggestIndex);
        });
        CompletableFuture<Void> albumIntroCompletableFuture = CompletableFuture.runAsync(() -> {
            SuggestIndex albumIntroSuggestIndex = new SuggestIndex();
            albumIntroSuggestIndex.setId(UUID.randomUUID().toString().replaceAll("-", ""));
            albumIntroSuggestIndex.setTitle(albumInfoIndex.getAlbumIntro());
            albumIntroSuggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumIntro()}));
            albumIntroSuggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumIntro())}));
            albumIntroSuggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumIntro())}));
            this.suggestIndexRepository.save(albumIntroSuggestIndex);
        });
        CompletableFuture<Void> announceNameCompletableFuture = CompletableFuture.runAsync(() -> {
            SuggestIndex announcerSuggestIndex = new SuggestIndex();
            announcerSuggestIndex.setId(UUID.randomUUID().toString().replaceAll("-", ""));
            announcerSuggestIndex.setTitle(albumInfoIndex.getAnnouncerName());
            announcerSuggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAnnouncerName()}));
            announcerSuggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAnnouncerName())}));
            announcerSuggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAnnouncerName())}));
            suggestIndexRepository.save(announcerSuggestIndex);
        });

        double hotScore = new Random().nextInt(100);
        albumInfoIndex.setHotScore(hotScore);
        CompletableFuture.allOf(
                albumInfoCompletableFuture,
                statCompletableFuture,
                categoryCompletableFuture,
                userCompletableFuture,
                keyWordCompletableFuture,
                attributeCompletableFuture,
                albumIntroCompletableFuture,
                announceNameCompletableFuture
        ).join();

        RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumInfoIndex.getId());
        albumIndexRepository.save(albumInfoIndex);


    }
}
